package kafkaUtil

import (
	"context"
	"fmt"
	"github.com/segmentio/kafka-go"
	"log"
	"os/exec"
	"strings"
	"time"
)

// 全局资源（实际项目中建议用依赖注入）
var (
	KafkaWriter  *kafka.Writer
	KafkaBrokers = []string{"127.0.0.1:9092"} // Kafka broker地址列表
	KafkaScript  = "kafka-topics"             // 替换为你的kafka-topics.sh路径, 例如:/opt/kafka/bin/kafka-topics.sh
)

// 初始化Kafka生产者（含Topic自动检测）
func InitKafka() error {
	// 先测试连接
	if err := TestKafkaConnection(); err != nil {
		return fmt.Errorf("kafka连接失败: %v", err)
	}

	// 可以通过命令行 kafka-topics --list --bootstrap-server localhost:9092 查看当前的topic

	// 初始化Writer
	KafkaWriter = &kafka.Writer{
		Addr: kafka.TCP(KafkaBrokers[0]),
		//Topic:        defaultTopic, // 动态指定主题
		Balancer:     &kafka.LeastBytes{},
		RequiredAcks: kafka.RequireAll, // 单机环境用RequireOne，集群用RequireAll
		MaxAttempts:  3,                // 重试次数
	}
	log.Println("Kafka连接成功")

	return nil
}

// 检查Topic是否存在（通过命令行工具）
func CheckTopicExists(topic string) (bool, error) {
	// 执行命令查询Topic列表
	cmd := exec.Command(
		KafkaScript,
		"--list",
		"--bootstrap-server", KafkaBrokers[0],
	)
	output, err := cmd.CombinedOutput()
	if err != nil {
		return false, fmt.Errorf("查询Topic列表失败: %v, 输出: %s", err, string(output))
	}

	// 检查输出中是否包含目标Topic
	return strings.Contains(string(output), topic), nil
}

// 创建Topic（通过命令行工具）
func CreateTopic(topic string) error {
	cmd := exec.Command(
		KafkaScript,
		"--create",
		"--topic", topic,
		"--bootstrap-server", KafkaBrokers[0],
		"--partitions", "3",
		"--replication-factor", "1", // 单机用1，集群≥2
	)
	output, err := cmd.CombinedOutput()
	if err != nil {
		// 忽略"已存在"错误
		if strings.Contains(string(output), "already exists") {
			return nil
		}
		return fmt.Errorf("创建Topic失败: %v, 输出: %s", err, string(output))
	}
	return nil
}
func TestKafkaConnection() error {
	// 创建临时连接测试
	conn, err := kafka.Dial("tcp", "127.0.0.1:9092")
	if err != nil {
		return err
	}
	defer conn.Close()

	// 获取broker列表，验证集群状态
	brokers, err := conn.Brokers()
	if err != nil {
		return err
	}

	if len(brokers) == 0 {
		return fmt.Errorf("未找到可用的Kafka broker")
	}

	log.Printf("发现 %d 个Kafka broker", len(brokers))
	return nil
}

// 关闭Kafka生产者
func CloseKafka() {
	_ = KafkaWriter.Close()
}

// 发送Kafka消息（含Topic自动创建）
func SendKafkaMessage(ctx context.Context, topic, key, value string) error {
	// 检查Topic是否存在
	exists, err := CheckTopicExists(topic)
	if err != nil {
		return fmt.Errorf("检查Topic失败: %v", err)
	}
	// 不存在则创建
	if !exists {
		if err := CreateTopic(topic); err != nil {
			return fmt.Errorf("创建Topic失败: %v", err)
		}
		time.Sleep(2 * time.Second)
	}

	// 发送消息
	return KafkaWriter.WriteMessages(ctx, kafka.Message{
		Topic: topic,
		Key:   []byte(key),
		Value: []byte(value),
	})
}
